PostgreSQL last N in-table cache 探討

前些天對悠遊卡儲值時,加值機有提供一個最近六次的紀錄查詢功能.
類似這種功能,在電商查詢最近消費紀錄等等,都帶給消費者便利.
進而想以這個功能來做以下的探討.

建立測試資料

create table mrt_cstmr (
  cardid text not null
);

insert into mrt_cstmr values
('A123'),('A456'),('A789');


create table mrt_tap_dtl (
  cardid text not null
, ts timestamp not null
, stname text not null
, primary key (cardid, ts)
);

insert into mrt_tap_dtl
select c.cardid
     , '2021-01-01 06:00:00'::timestamp 
     + interval '1 day' * n 
     + interval '1 minute' * ceil(random() * 1080)::int
     , (array['新店','七張', '景美', '萬隆'
        , '公館', '古亭', '西門', '北門'
        , '中山', '松山', '象山', '大安'
        ]::text[])[ceil(random() * 12)]
  from mrt_cstmr c
     , generate_series(1, 8) as g(n);

使用 Window function 來取出最後 6 筆

-- 取最後六筆
select *
  from (select *
             , rank() over(partition by cardid order by ts desc) as rnk
          from mrt_tap_dtl) a
 where rnk <= 6;

-- 查詢計畫可用以下語法
explain (analyze, verbose, costs, buffers, timing, summary)
select *
  from (select *
             , rank() over(partition by cardid order by ts desc) as rnk
          from mrt_tap_dtl) a
 where rnk <= 6;

--上面是很標準的步驟,單一使用查詢時語法如下

explain (analyze, verbose, costs, buffers, timing, summary)
select *
  from (select *
             , rank() over(partition by cardid order by ts desc) as rnk
          from mrt_tap_dtl
         where cardid = 'A123') a
 where rnk <= 6;

基本方法的探討

需要進行排序(sort).這需要消耗資源,增加執行的時間,隨著資料量增大, 消耗的資源會相對增加.
為了控制資源消耗,保持適當的反應速度,常見的方式可將歷史資料移到另外的table 或是 database,線上的table 維持一季度或是數個月的資料.
但是消費者上次消費或操作可能是去年或更早,若能夠將最近N筆的紀錄,適當保留,類似 cache 的概念,可以提供更好的服務.甚至可以減少對紀錄表(如上面例子中 mrt_tap_dtl)的查詢, 紀錄表只做insert , 進而減少 lock 等資源消耗.

為了要做此類似 cache 的功能,我們需要建立另一個 table, 將每個卡號的最後消費或操作紀錄保存.先做最後一筆,後面再來進化.
這時候有兩種常用的方式可以選擇,一種是 insert 到記錄表時,也對 cache 表做相關操作.
因為 cache 表只保留最新的,所以當cardid 是第一次進入 cache表時,要做 insert,後續操作為 update.
這樣的方式是直覺,但是每個對紀錄表做 insert 的程式,都需要確保進行兩段操作,若有程式遺漏對 cache 表操作,就容易產生不一致.
另一種方式是使用trigger.由資料庫的機制來確保,避免遺漏.

至於insert 與 update 結合的語法,在不同資料庫,語法略有差異.

MySQL 部分可以參考
https://dev.mysql.com/doc/refman/5.7/en/insert-on-duplicate.html

PostgreSQL
https://docs.postgresql.tw/reference/sql-commands/insert

裡面的 ON CONFLICT Clause

https://docs.postgresql.tw/reference/sql-commands/insert#on-conflict-clause

使用 trigger 來紀錄最後一次操作

--先建立 cache 用途的 table

create table mrt_tap_cach (
  cardid text not null primary key
, last_stname text not null
, last_ts timestamp not null
);

--接著來寫 trigger function

create or replace function mrt_tap_dtl_insert()
returns trigger
language plpgsql
as $$
begin
  insert into mrt_tap_cach(cardid, last_stname, last_ts)
  values (new.cardid, new.stname, new.ts)
  on conflict (cardid) do update
  set last_stname = new.stname
    , last_ts = new.ts
  ;
  return new;
end;
$$;


--接著建立 trigger

create trigger mrt_tap_dtl_insert_trig
after insert on mrt_tap_dtl
for each row
execute procedure mrt_tap_dtl_insert();

接著來做測試

insert into mrt_tap_dtl
select c.cardid
     , '2021-02-01 06:00:00'::timestamp 
     + interval '1 day' * n 
     + interval '1 minute' * ceil(random() * 1080)::int
     , (array['新店','七張', '景美', '萬隆'
        , '公館', '古亭', '西門', '北門'
        , '中山', '松山', '象山', '大安'
        ]::text[])[ceil(random() * 12)]
  from mrt_cstmr c
     , generate_series(1, 3) as g(n); 

--這樣會輸入 9 筆.(3個卡號,每個 3 筆).

--先來看 mrt_tap_dtl,裡面新的資料

select *
  from (select *
             , rank() over(partition by cardid order by ts desc) as rnk
          from mrt_tap_dtl) a
 where rnk <= 3;

 cardid |         ts          | stname | rnk
--------+---------------------+--------+-----
 A123   | 2021-02-04 12:59:00 | 古亭   |   1
 A123   | 2021-02-03 14:54:00 | 公館   |   2
 A123   | 2021-02-02 08:34:00 | 七張   |   3
 A456   | 2021-02-04 20:37:00 | 七張   |   1
 A456   | 2021-02-03 14:50:00 | 中山   |   2
 A456   | 2021-02-02 23:21:00 | 象山   |   3
 A789   | 2021-02-04 10:15:00 | 公館   |   1
 A789   | 2021-02-03 15:06:00 | 古亭   |   2
 A789   | 2021-02-02 20:21:00 | 北門   |   3
(9 rows)

--再來看 mrt_tap_cach 保留的最後操作資料.

select *
  from mrt_tap_cach;

 cardid | last_stname |       last_ts
--------+-------------+---------------------
 A123   | 古亭        | 2021-02-04 12:59:00
 A456   | 七張        | 2021-02-04 20:37:00
 A789   | 公館        | 2021-02-04 10:15:00
(3 rows)

有效的紀錄了!
至此並無太多新的技術,保留最後一筆的功能 trigger 也是屬於廣泛應用.

擴展紀錄空間的探討

接著來思考一下,怎樣保留最後 N 筆,在此先以 6 筆為例.
目前在 mrt_tap_cach 中,除了卡號,我們建立了兩個欄位,保留站名與時間戳.
若要保留6筆,是否再建立 10 個欄位,也就是 5 組, 但是怎樣做 推陳出新 ??
這類操作對 SQL 語法來說並不自然.
是否有什麼好的方式,提供一個資料結構,來做推陳出新.

首先我們來看 PostgreSQL 是有 array 的資料型態.
可以設法使用 array 來存放,再做推陳出新的操作.
另外我們也知道 PostgreSQL可以使用 PL/Python.
可以用 PL/Python 來寫 function 與 trigger.
透過 array 可以將資料傳給 PL/Python 的函數, 再利用 Python 處理 List
強大的功能.看來似乎有點眉目了.
另外查找 Python 的函數庫, 在 collections 中有一個 deque

https://docs.python.org/zh-tw/3/library/collections.html#collections.deque

搭配使用,就可以很方便的做限制長度,而且有推陳出新的功能,無需重新造輪子.相關搭配技術似乎找齊了.

摻在一起吧!

首先我們將 mrt_tap_cach 增加一個 text[] ,也就是text型態的 array .

alter table mrt_tap_cach
add column last_n text[];

--接著來寫 trigger function. 因為 mrt_tap_dtl_insert_trig 呼叫的
--mrt_tap_dtl_insert() 使用了 insert on conflict, 實際上會有
--insert 與 update 兩種操作.

create or replace function mrt_tap_cach_insert_update()
returns trigger
language plpython3u
as $$
  from collections import deque
  d = deque(maxlen = 6)
  if TD['new']['last_n'] != None:
    for elem in TD['new']['last_n']:
      d.append(elem)
  d.appendleft([TD['new']['last_stname'], TD['new']['last_ts']])
  TD['new']['last_n'] = list(d)
  rv = 'MODIFY'
  return rv
$$;

--建立兩個 trigger

create trigger mrt_tap_cach_insert_trig
before insert on mrt_tap_cach
for each row
execute procedure mrt_tap_cach_insert_update();

create trigger mrt_tap_cach_update_trig
before update on mrt_tap_cach
for each row
execute procedure mrt_tap_cach_insert_update();

測試!

--先將 mrt_tap_dtl 與 mrt_tap_cach 的資料 truncate 

truncate table mrt_tap_dtl;
truncate table mrt_tap_cach;

insert into mrt_tap_dtl
select c.cardid
     , '2021-03-01 06:00:00'::timestamp 
     + interval '1 day' * n 
     + interval '1 minute' * ceil(random() * 1080)::int
     , (array['新店','七張', '景美', '萬隆'
        , '公館', '古亭', '西門', '北門'
        , '中山', '松山', '象山', '大安'
        ]::text[])[ceil(random() * 12)]
  from mrt_cstmr c
     , generate_series(1, 3) as g(n); 
     
--來看一下 mrt_tap_cach 的內容

select *
  from mrt_tap_cach;

-[ RECORD 1 ]-----------------------------------------------------------------------------------------
cardid      | A123
last_stname | 北門
last_ts     | 2021-03-04 20:34:00
last_n      | {{北門,"2021-03-04 20:34:00"},{中山,"2021-03-03 20:47:00"},{七張,"2021-03-02 13:53:00"}}
-[ RECORD 2 ]-----------------------------------------------------------------------------------------
cardid      | A456
last_stname | 新店
last_ts     | 2021-03-04 19:20:00
last_n      | {{新店,"2021-03-04 19:20:00"},{萬隆,"2021-03-03 13:23:00"},{公館,"2021-03-02 11:16:00"}}
-[ RECORD 3 ]-----------------------------------------------------------------------------------------
cardid      | A789
last_stname | 北門
last_ts     | 2021-03-04 17:41:00
last_n      | {{北門,"2021-03-04 17:41:00"},{新店,"2021-03-03 06:30:00"},{萬隆,"2021-03-02 23:31:00"}}

如同預期,再來每個 cardid 增加 4 筆,觀察長度限制.

insert into mrt_tap_dtl
select c.cardid
     , '2021-03-04 06:00:00'::timestamp 
     + interval '1 day' * n 
     + interval '1 minute' * ceil(random() * 1080)::int
     , (array['新店','七張', '景美', '萬隆'
        , '公館', '古亭', '西門', '北門'
        , '中山', '松山', '象山', '大安'
        ]::text[])[ceil(random() * 12)]
  from mrt_cstmr c
     , generate_series(1, 4) as g(n); 
     
select *
  from mrt_tap_cach;
  
...
-[ RECORD 3 ]--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
cardid      | A789
last_stname | 北門
last_ts     | 2021-03-08 23:47:00
last_n      | {{北門,"2021-03-08 23:47:00"},{象山,"2021-03-07 15:31:00"},{新店,"2021-03-06 20:52:00"},{大安,"2021-03-05 15:17:00"},{北門,"2021-03-04 17:41:00"},{新店,"2021-03-03 06:30:00"}}

--保留到 {新店,"2021-03-03 06:30:00"} 而 {萬隆,"2021-03-02 23:31:00"} 已經被推出去了!

至此已經達到保留最後 6 筆的操作紀錄的功能.若想要增加保留的筆數,只要修改 mrt_tap_cach_insert_update()
trigger function 即可,也不需要重新建立trigger,也不用修改 table, 影響極小.

create or replace function mrt_tap_cach_insert_update()
returns trigger
language plpython3u
as $$
  from collections import deque
  d = deque(maxlen = 8)
  if TD['new']['last_n'] != None:
    for elem in TD['new']['last_n']:
      d.append(elem)
  d.appendleft([TD['new']['last_stname'], TD['new']['last_ts']])
  TD['new']['last_n'] = list(d)
  rv = 'MODIFY'
  return rv
$$;


insert into mrt_tap_dtl
select c.cardid
     , '2021-03-15 06:00:00'::timestamp 
     + interval '1 day' * n 
     + interval '1 minute' * ceil(random() * 1080)::int
     , (array['新店','七張', '景美', '萬隆'
        , '公館', '古亭', '西門', '北門'
        , '中山', '松山', '象山', '大安'
        ]::text[])[ceil(random() * 12)]
  from mrt_cstmr c
     , generate_series(1, 9) as g(n); 
     
select *
  from mrt_tap_cach;

...
-[ RECORD 3 ]-----------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------------------------------------------------
cardid      | A789
last_stname | 象山
last_ts     | 2021-03-24 13:18:00
last_n      | {{象山,"2021-03-24 13:18:00"},{象山,"2021-03-23 16:40:00"},{西門,"2021-03-22 13:41:00"},{北門,"2021-03-21 21:08:00"},{古亭,"2021-03-20 07:44:00"},{公館,"2021-03-19 13:11:00"},{景美,"2021-03-18 17:26:00"},{大安,"2021-03-17 09:10:00"}}

以上是綜合運用了 trigger , plpython, array 等功能,來達到一個簡單輕量的 last n in-table cache 功能.

結語

trigger 或是 plpython 結合運用,可以擴展使用彈性.進而減少對table的查詢,提高整體的效能.